package com.shujia.city.dwd

import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object DwdGjjSspersonsMskDay extends SparkTool {
  /**
   * 在run函数中写数据处理的逻辑
   * import spark.implicits._
   * import org.apache.spark.sql.functions._
   */
  override def run(spark: SparkSession, ds: String): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._

    //1、读取数据
    val sspersons: Dataset[Row] = spark
      .table("ods.ods_t_gjj_sspersons")
      .where($"ds" === ds)

    //2、处理数据
    val resultDF: DataFrame = sspersons.select(
      $"spcode",
      $"hjstatus",
      $"sncode",
      upper(md5(concat($"spname", expr("'shujia'")))) as "spname",
      upper(md5(concat($"id", expr("'shujia'")))) as "id",
      upper(md5(concat($"spcard", expr("'shujia'")))) as "spcard",
      upper(md5(concat($"sppassword", expr("'shujia'")))) as "sppassword",
      $"zjfdm",
      $"spkhrq",
      $"spperm",
      $"spgz",
      $"spsingl",
      $"spjcbl",
      $"spmfact",
      $"spmfactzg",
      $"spjym",
      $"ncye",
      $"splast",
      $"dwbfye",
      $"grbfye",
      $"spmend",
      $"splastlx",
      $"spout",
      $"spin",
      $"bnlx",
      $"nclx",
      $"dwhjny",
      $"zghjny",
      $"btyje",
      $"btye",
      $"btbl",
      $"bthjny",
      $"spxh",
      $"spzy",
      $"spxhrq",
      $"splr",
      $"spoldbankno",
      $"spdk",
      $"spdy",
      $"zhdj",
      $"spnote",
      $"modifytime",
      $"status",
      $"cbank",
      $"bcyje",
      $"bcye",
      $"bcbl",
      $"bchjny",
      $"zjzl"
    )
    //保存数据
    saveToTable(spark, resultDF, "dwd.dwd_gjj_sspersons_msk_d", ds)

  }
}
